"""Utility helpers for feature quality analysis."""

from __future__ import annotations

from collections import defaultdict
from typing import Dict, Iterable

import numpy as np
import pandas as pd

# 定义功能重叠特征组（同一组内仅保留 1-2 个代表特征）
FUNCTIONAL_OVERLAP_GROUPS: list[set[str]] = [
    {"rsi_14", "williams_r_14", "cmo_14"},
    {"macd_dif", "macd_dea", "macd_hist"},
]

# 特征类别映射用于覆盖率统计
CATEGORY_PATTERNS: Dict[str, Iterable[str]] = {
    "price_trend": ("sma", "ema", "bias", "macd", "trend", "close_lag", "price_position"),
    "volume": ("volume", "obv"),
    "volatility": ("volatility", "atr", "realized_volatility", "variance"),
    "technical_indicator": (
        "rsi",
        "cmo",
        "williams",
        "kdj",
        "cci",
        "adx",
        "mfi",
    ),
    "interaction": ("interaction", "ratio"),
}


def compute_coefficient_of_variation(series: pd.Series) -> float | None:
    """计算变异系数（标准差/均值），均值接近 0 时返回 None."""
    clean = series.dropna()
    if clean.empty:
        return None
    mean_val = clean.mean()
    if abs(mean_val) < 1e-9:
        return None
    std_val = clean.std()
    if pd.isna(std_val):
        return None
    return float(std_val / abs(mean_val))


def compute_iqr_outlier_ratio(series: pd.Series) -> float:
    """计算 IQR 异常值占比."""
    clean = series.dropna()
    if clean.empty:
        return 0.0
    q1 = clean.quantile(0.25)
    q3 = clean.quantile(0.75)
    iqr = q3 - q1
    if iqr == 0:
        return 0.0
    lower = q1 - 1.5 * iqr
    upper = q3 + 1.5 * iqr
    mask = (clean < lower) | (clean > upper)
    return float(mask.sum() / len(series))


def winsorize_series(series: pd.Series, lower_quantile: float = 0.01, upper_quantile: float = 0.99) -> pd.Series:
    """对序列执行 Winsorize 裁剪."""
    if series.dropna().empty:
        return series.copy()
    lower = series.quantile(lower_quantile)
    upper = series.quantile(upper_quantile)
    return series.clip(lower=lower, upper=upper)


def categorize_feature(name: str) -> str:
    """根据名称推断特征类别."""
    lname = name.lower()
    for category, patterns in CATEGORY_PATTERNS.items():
        if any(pattern in lname for pattern in patterns):
            return category
    return "other"


def summarize_feature_categories(features: Iterable[str]) -> Dict[str, int]:
    """统计特征类别分布."""
    summary: Dict[str, int] = defaultdict(int)
    for feat in features:
        summary[categorize_feature(feat)] += 1
    return dict(summary)

